package com.nami.gchat.domain.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.github.yitter.idgen.YitIdHelper;
import com.google.protobuf.ByteString;
import com.nami.gate.external.basic.bo.ConnectedUserBO;
import com.nami.gate.external.basic.cache.user.UserDeviceRel;
import com.nami.gate.external.basic.event.EventPublisher;
import com.nami.gate.external.basic.event.enums.EventType;
import com.nami.gate.external.basic.event.factory.EventFactory;
import com.nami.gchat.domain.model.co.GrpInfoModel;
import com.nami.gchat.domain.model.co.GrpMemberModel;
import com.nami.gchat.domain.repository.CacheDomainRepository;
import com.nami.gchat.domain.service.GrpChatDomainService;
import com.nami.gchat.types.utils.ConverUtil;
import com.nami.im.network.model.message.v2.Message;
import com.nami.transport.proto.codec.MqTransMsg;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.dromara.common.core.enums.user.UserStatusEnum;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import static com.nami.gchat.types.constants.GrpConst.COMMON_SMALL_GRP_MEMBER_COUNT;

@Slf4j
@Component
@RequiredArgsConstructor
public class GrpChatDomainServiceImpl implements GrpChatDomainService {

    private final EventPublisher eventPublisher;
    private final CacheDomainRepository cacheDomainRepository;

    @Override
    public boolean validateUserPermission(UserDeviceRel user) {
        if (user == null) {
            return Boolean.TRUE;
        }
        if (checkUserBanned(user)) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }

    @Override
    public boolean validateGrpParams(GrpInfoModel grpInfo) {
        if (grpInfo == null) {
            return Boolean.TRUE;
        } // end check group null if
        // 对群校验
        if (grpInfo.getBanSpeak() != null && grpInfo.getBanSpeak()) {
            return Boolean.TRUE;
        } // end banspeak check
        if (!grpInfo.getAppKey().equals(grpInfo.getAppKey())) {
            return Boolean.TRUE;
        } // end appkey check
        return Boolean.FALSE;
    }

    @Override
    public boolean validateMePermission(GrpMemberModel me) {
        if (me == null) {
            return Boolean.TRUE;
        }
        if (me.getBanSpeak()) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }

    @Override
    public boolean validateBasicParams(ConnectedUserBO sender, Message message, String targetId) {
        if (checkParamsNull(sender, message, targetId)) {
            return Boolean.TRUE;
        }
        // senderId 从gate绑定的userId 取出来，不从客户端取所以不用校验
//        if (checkSameUser(sender, message)) {
//            log.warn("[c2g] 校验到乱发消息，或参数为null. clientMsgId={}", message.getClientMsgId());
//            return Boolean.TRUE;
//        }
        return Boolean.FALSE;
    }

    /**
     *                          群聊消息
     *                              │
     *                              ▼
     *                    +-------------------+
     *                    |   分片决策层       |
     *                    | (判断分片逻辑)     |
     *                    +-------------------+
     *                              │
     *          +-------------------+-------------------+
     *          ▼                   ▼                   ▼
     * +------------------+ +------------------+ +------------------+
     * | 小群服务          | | 大群服务          | | 超大群服务        |
     * | (Hash路由)        | | (动态弹性分片)    | | (边缘分片+推拉)   |
     * +------------------+ +---------+--------+ +------------------+
     *                                │
     *                                ▼
     *                    +-----------------------+
     *                    | 分片执行层            |
     *                    | - 按在线用户动态分片   |
     *                    | - 虚拟分片共享线程池   |
     *                    +-----------------------+
     *
     * 群聊下发服务分层：
     * ├── 小群服务（<1000人）
     * ├── 大群服务（1000~1万人） → v1 拆分为6个分片（3000人/500人分片）-> v2 动态拆分
     * └── 超大群服务（>1万人） → v1 拆为200个分片（10万人/500人分片）-> v2 动态拆分
     *
     * @param sender
     * @param message
     * @param allMembers
     * @return
     */

    @Override
    public boolean publishMessageEvent(Integer qos,
                                       ConnectedUserBO sender,
                                       Message message,
                                       List<GrpMemberModel> allMembers) {
        MqTransMsg.GrpLogic2DispatchBody trans = newMqTransBody(qos, sender, message, allMembers);
        // 存储消息
        eventPublisher.publish(EventFactory.newEvent(EventType.STORE_GRP_MSG, trans));
        // 更新会话
        eventPublisher.publish(EventFactory.newEvent(EventType.UPD_GRP_CONVERS, trans, message.getTargetId()));

        // 下发消息
        // 动态分片+多级队列，通过计算用户在线状态和负载情况动态选择下发节点。
        // TODO 按照规模进行拆分
        // date 2025-02-19
        // 500人 普通群，消息下发，采用grpId hash下发
        // 501-3001人 大群 采用按照 hash+分片 处理
        // 超大群，走独立的下发服务topic, 专门处理超大群

        // v1
        // 按照聊天类型拆分为 ===> 单聊，群聊下发服务 分别处理。
        // 群聊下发服务，再次拆分为，小群下发服务，大群下发服务，超大群下发服务。
        // 大群以上再次拆分 ===> 按片拆分，将用户拆分，比如3000人群，将获取到的群成员拆分为6个分片，即拆分为小群规模，不同实例处理的“在线”用户规模就变得很小了
        // 超大群 10w人群，拆为200个分片。


        //
        // 1. 超大群 消费者，上线就往redis注册，服务下线就把数量-1
        // 2. 消息设置为广播，每个消费者都接收到消息后，计算自己的分片进行消费。
        // 3. 群成员分片简单拆分，不按照在线进行拆分。离线用户可能还需发送推送消息


        // v2 动态分片
        // 按照聊天类型拆分为 ===> 单聊，群聊下发服务 分别处理。
        // 群聊下发服务，再次拆分为，小群下发服务，大群下发服务，超大群下发服务。
        //   更新用户在线状态
        //void onUserOnline(String userId, String groupId) {
        //    redis.sadd("group_online:" + groupId, userId);
        //}

        // 根据在线人数和消息频率调整分片数。
        //int onlineCount = redis.scard("group_online:" + groupId);
        //int shardCount = Math.max(onlineCount / 500, currentQPS / 1000);

        // 虚拟分片映射
        // 在线用户哈希到虚拟分片，物理实例轮询消费虚拟分片。
        // 分片键 = groupId + 虚拟分片ID
        //String shardKey = groupId + "_" + (hash(userId) % shardCount);
        //rocketMQTemplate.sendOrderly("group_topic", message, shardKey);

        //超大群边缘分片
        //根据用户地理位置，将分片分配到最近的边缘节点（如阿里云ENS）。
        // 按地理位置选择边缘分片
        //String edgeRegion = geoLocator.getRegion(userIp);
        //String shardKey = groupId + "_" + edgeRegion;

        /**
         *
         * 大群每个分片处理数： 读取yml, 配置注解，做到实时更新参数
         *                 int shardCount = 获取群成员数量 / 分片处理数
         *
         * for (int i=0; i< shardCount; i++) {
         *   rocketmq.sendOrderly("big_group", body,i);
         * }
         *
         *
         */

        // 死循环遍历队列 消费即可！
        // 队列使用优先级队列
        // 将分发分为 高中低 不同优先级队列。
        // 按照优先级进行消费

        eventPublisher.publish(EventFactory.newEvent(EventType.DISPATCH_GRP_MSG, trans, message.getTargetId()));
        return false;
    }

    @Override
    public boolean checkIdempotent(String appKey, String senderId, String clientMsgId) {
        return cacheDomainRepository.checkIdempotent(appKey, senderId, clientMsgId);
    }

    @Override
    public void fillParam(String appKey, Message message) {
        // seq
        Long maxSeq = cacheDomainRepository.incrSeq(appKey,
            ConverUtil.genGrpCid(message.getTargetId()));

        message.setMsgSeqNo(maxSeq);
        message.setMsgId(YitIdHelper.nextId() + "");
    }

    private static MqTransMsg.GrpLogic2DispatchBody newMqTransBody(Integer qos,
                                                                   ConnectedUserBO sender,
                                                                   Message message,
                                                                   List<GrpMemberModel> allMembers) {
        //membersExcludeMe
        Set<String> allMemberUserIds = allMembers.stream()
//            .filter(item -> !message.getSenderId().equals(item.getUserId()))
            .map(GrpMemberModel::getUserId)
            .collect(Collectors.toSet());

        if (CollUtil.isNotEmpty(allMembers) && allMemberUserIds.size() > COMMON_SMALL_GRP_MEMBER_COUNT) {
            allMemberUserIds = Collections.EMPTY_SET;
        }

        MqTransMsg.TransUser senderUser = MqTransMsg.TransUser.newBuilder()
            .setUserId(sender.getUserId())
            .setPlatform(sender.getPlatform())
            .setAppKey(sender.getAppKey())
            .setConnectId(sender.getConnectId())
            .setInstanceId(sender.getInstanceId())
            .build();

        MqTransMsg.GrpLogic2DispatchBody request = MqTransMsg.GrpLogic2DispatchBody.newBuilder()
            .setSuccess(Boolean.TRUE)
            .setMessage(ByteString.copyFromUtf8(JsonUtils.toJsonString(message)))
            .setSender(senderUser)
            .addAllReceiverIds(allMemberUserIds)
            .setMemberCnt(allMemberUserIds.size())
            .setQos(qos)
            .build();

        return request;
    }

    public boolean checkParamsNull(ConnectedUserBO sender, Message message, String targetId) {
        if (sender == null
            || StringUtils.isBlank(sender.getPlatform())
            || message == null
            || StringUtils.isBlank(message.getMsgContent())
            || StringUtils.isBlank(sender.getUserId())
            || StringUtils.isBlank(targetId)
            || StringUtils.isBlank(message.getClientUid())) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }

//    public boolean checkSameUser(ConnectedUserCO sender, Message message) {
//        if (StringUtils.isBlank(sender.getUserId()) || !message.getSenderId().equals(sender.getUserId())
//            || StringUtils.isBlank(sender.getAppKey())) {
//            return Boolean.TRUE;
//        }
//        return Boolean.FALSE;
//    }


    private static boolean checkUserBanned(UserDeviceRel user) {
        if (UserStatusEnum.BAN.getCode() == user.getStatus()) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }


}
